package cn._51doit.day02;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 标准键盘布局
 * ┌───┐   ┌───┬───┬───┬───┐ ┌───┬───┬───┬───┐ ┌───┬───┬───┬───┐ ┌───┬───┬───┐
 * │Esc│   │ F1│ F2│ F3│ F4│ │ F5│ F6│ F7│ F8│ │ F9│F10│F11│F12│ │P/S│S L│P/B│  ┌┐    ┌┐    ┌┐
 * └───┘   └───┴───┴───┴───┘ └───┴───┴───┴───┘ └───┴───┴───┴───┘ └───┴───┴───┘  └┘    └┘    └┘
 * ┌───┬───┬───┬───┬───┬───┬───┬───┬───┬───┬───┬───┬───┬───────┐ ┌───┬───┬───┐ ┌───┬───┬───┬───┐
 * │~ `│! 1│@ 2│# 3│$ 4│% 5│^ 6│& 7│* 8│( 9│) 0│_ -│+ =│ BacSp │ │Ins│Hom│PUp│ │N L│ / │ * │ - │
 * ├───┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─────┤ ├───┼───┼───┤ ├───┼───┼───┼───┤
 * │ Tab │ Q │ W │ E │ R │ T │ Y │ U │ I │ O │ P │{ [│} ]│ | \ │ │Del│End│PDn│ │ 7 │ 8 │ 9 │   │
 * ├─────┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴─────┤ └───┴───┴───┘ ├───┼───┼───┤ + │
 * │ Caps │ A │ S │ D │ F │ G │ H │ J │ K │ L │: ;│" '│ Enter  │               │ 4 │ 5 │ 6 │   │
 * ├──────┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴────────┤     ┌───┐     ├───┼───┼───┼───┤
 * │ Shift  │ Z │ X │ C │ V │ B │ N │ M │< ,│> .│? /│  Shift   │     │ ↑ │     │ 1 │ 2 │ 3 │   │
 * ├─────┬──┴─┬─┴──┬┴───┴───┴───┴───┴───┴──┬┴───┼───┴┬────┬────┤ ┌───┼───┼───┐ ├───┴───┼───┤ E││
 * │ Ctrl│    │Alt │         Space         │ Alt│    │    │Ctrl│ │ ← │ ↓ │ → │ │   0   │ . │←─┘│
 * └─────┴────┴────┴───────────────────────┴────┴────┴────┴────┘ └───┴───┴───┘ └───────┴───┴───┘
 *
 * @create: 2021-10-15 10:04
 * @author: 今晚打脑斧
 * @program: SocketSource
 * @Description:
 *    1.说明程序的并行度，决定了分布式程序执行的效率，但是也要考虑资源
 *    2.通过env的socketTextStream方法创建的DataStream的并行度是为1（只能为1，不能是多个）
 *         socketTextStream是一个非并行的Source
 *    3.调用完map方法的到的DataStream并行的为多个并行，如果没有指定，那么并行度与执行环境Env保持一致
 **/
public class SocketSource {
    public static void main(String[] args) throws Exception{
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //获取执行环境的并行数
        int parallelism = env.getParallelism();
        System.out.println("执行环境的并行度：" + parallelism);

        //调用Env的Source方法创建DataStream
        DataStreamSource<String> lines = env.socketTextStream("doit01", 8888);

        //获得socketTextStream创建的DataStreamSource的并行度
        int parallelism1 = lines.getParallelism();
        System.out.println("socketTextStream创建的DataStreamSource的并行度：" + parallelism1);

        //调用Transformation,将数据转大写,并获取并行度
        SingleOutputStreamOperator<String> uppered = lines.map(line -> line.toUpperCase());
        int parallelism2 = uppered.getParallelism();
        System.out.println("调用完map方法得到的DataStream的并行度：" + parallelism2);

        //sink 调用Sink将数据输出
        DataStreamSink<String> print = uppered.print();

        //获得Sink的并行度
        int parallelism3 = print.getTransformation().getParallelism();
        System.out.println("调用完print方法得到的DataStreamSink的并行度：" + parallelism3);

        //启动
        env.execute();
    }
}
